package com.markhsiu.MiniMQ.broker.task;

import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;

import com.markhsiu.MiniMQ.broker.factory.BrokerFactory;
import com.markhsiu.MiniMQ.broker.store.StoreManager;
import com.markhsiu.MiniMQ.channel.ChannelProcessor;
import com.markhsiu.MiniMQ.model.Message;
import com.markhsiu.MiniMQ.model.constant.MessageCmdEnum;
import com.markhsiu.MiniMQ.model.constant.MessageSourceEnum;

public class MessagePushTask implements Callable<Void>{
 
	private BrokerFactory broker = BrokerFactory.instance();
	
	public  MessagePushTask() {}
	
	@Override
	public Void call() throws Exception {
		
		while(true){
			Map<String,List<String>> consumers  = broker.getAllConsumers();
			for(Map.Entry<String, List<String>> consumer : consumers.entrySet()){
				ChannelProcessor channel = broker.getConsumerConnection(consumer.getKey());
				for (String topic : consumer.getValue()) {
					Message message = StoreManager.pollMessage(topic);
					message.setSource(MessageSourceEnum.BROKER);
					message.setTarget(MessageSourceEnum.CUSTOMER);
					message.setCmd(MessageCmdEnum.NEW);
					channel.writeAndFlush(message);
				}
			}
		}
	}

}
